package com.seedog.mq;

import com.seedog.mq.model.Message;
import com.seedog.mq.core.MessageQueue;
import com.seedog.mq.exception.TopicNotFoundException;
import com.seedog.mq.exception.ConsumerException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.UUID;

/**
 * 消费者实现类，提供从消息队列接收消息的功能
 */
public class ConsumerImpl implements Consumer {
    private static final Logger logger = LoggerFactory.getLogger(ConsumerImpl.class);
    
    private final String consumerId;
    private final MessageQueue messageQueue;
    private final Set<String> subscribedTopics;
    private final AtomicBoolean closed = new AtomicBoolean(false);
    
    public ConsumerImpl(MessageQueue messageQueue) {
        if (messageQueue == null) {
            throw new IllegalArgumentException("MessageQueue cannot be null");
        }
        
        this.consumerId = "consumer-" + UUID.randomUUID().toString();
        this.messageQueue = messageQueue;
        this.subscribedTopics = ConcurrentHashMap.newKeySet();
        logger.info("Consumer created with ID: {}", consumerId);
    }
    
    public ConsumerImpl(MessageQueue messageQueue, String consumerId) {
        if (messageQueue == null) {
            throw new IllegalArgumentException("MessageQueue cannot be null");
        }
        
        if (consumerId == null || consumerId.isEmpty()) {
            throw new IllegalArgumentException("Consumer ID cannot be null or empty");
        }
        
        this.consumerId = consumerId;
        this.messageQueue = messageQueue;
        this.subscribedTopics = ConcurrentHashMap.newKeySet();
        logger.info("Consumer created with ID: {}", consumerId);
    }
    
    @Override
    public String getConsumerId() {
        return consumerId;
    }
    
    @Override
    public void subscribe(String topicName) throws TopicNotFoundException {
        checkClosed();
        
        if (topicName == null || topicName.isEmpty()) {
            throw new IllegalArgumentException("Topic name cannot be null or empty");
        }
        
        if (!messageQueue.topicExists(topicName)) {
            throw new TopicNotFoundException(topicName);
        }
        
        // 在MessageQueueImpl中，当消费者首次poll消息时会自动添加到主题的消费者列表中
        // 这里只记录消费者订阅的主题
        subscribedTopics.add(topicName);
        logger.info("Consumer {} subscribed to topic {}", consumerId, topicName);
    }
    
    @Override
    public boolean unsubscribe(String topicName) {
        checkClosed();
        
        if (topicName == null || topicName.isEmpty()) {
            throw new IllegalArgumentException("Topic name cannot be null or empty");
        }
        
        boolean removed = subscribedTopics.remove(topicName);
        if (removed) {
            logger.info("Consumer {} unsubscribed from topic {}", consumerId, topicName);
        }
        return removed;
    }
    
    @Override
    public Message poll(String topicName) throws TopicNotFoundException, ConsumerException {
        return poll(topicName, 0); // 默认不等待
    }
    
    @Override
    public Message poll(String topicName, long timeout) throws TopicNotFoundException, ConsumerException {
        checkClosed();
        
        if (topicName == null || topicName.isEmpty()) {
            throw new IllegalArgumentException("Topic name cannot be null or empty");
        }
        
        // 检查是否已订阅该主题
        if (!subscribedTopics.contains(topicName)) {
            // 自动订阅不存在的主题
            subscribe(topicName);
        }
        
        try {
            Message message = messageQueue.pollMessage(topicName, consumerId, timeout);
            if (message != null) {
                logger.debug("Consumer {} received message {} from topic {}", 
                        consumerId, message.getMessageId(), topicName);
            }
            return message;
        } catch (TopicNotFoundException e) {
            throw e;
        } catch (Exception e) {
            logger.error("Failed to poll message from topic {} for consumer {}", topicName, consumerId, e);
            throw new ConsumerException("Failed to poll message: " + e.getMessage(), e);
        }
    }
    
    @Override
    public boolean acknowledge(String topicName, String messageId) throws TopicNotFoundException, ConsumerException {
        checkClosed();
        
        if (topicName == null || topicName.isEmpty()) {
            throw new IllegalArgumentException("Topic name cannot be null or empty");
        }
        
        if (messageId == null || messageId.isEmpty()) {
            throw new IllegalArgumentException("Message ID cannot be null or empty");
        }
        
        try {
            boolean acknowledged = messageQueue.acknowledge(topicName, consumerId, messageId);
            if (acknowledged) {
                logger.debug("Consumer {} acknowledged message {} from topic {}", 
                        consumerId, messageId, topicName);
            } else {
                logger.warn("Consumer {} failed to acknowledge message {} from topic {}", 
                        consumerId, messageId, topicName);
            }
            return acknowledged;
        } catch (TopicNotFoundException e) {
            throw e;
        } catch (Exception e) {
            logger.error("Failed to acknowledge message {} from topic {} for consumer {}", 
                    messageId, topicName, consumerId, e);
            throw new ConsumerException("Failed to acknowledge message: " + e.getMessage(), e);
        }
    }
    
    @Override
    public void close() {
        if (closed.compareAndSet(false, true)) {
            // 清空订阅的主题
            subscribedTopics.clear();
            logger.info("Consumer {} closed", consumerId);
        }
    }
    
    // 检查消费者是否已关闭
    private void checkClosed() {
        if (closed.get()) {
            throw new IllegalStateException("Consumer '" + consumerId + "' is closed");
        }
    }
}